
package top.hjcwzx.wrench.mqtt.service.core;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import top.hjcwzx.wrench.mqtt.common.Constant;
import top.hjcwzx.wrench.mqtt.config.MqttConfigProperties;
import top.hjcwzx.wrench.mqtt.event.MqttConnectedEvent;
import top.hjcwzx.wrench.mqtt.event.MqttConnectionLostEvent;
import top.hjcwzx.wrench.mqtt.service.publish.MqttEventPublisher;
import top.hjcwzx.wrench.mqtt.service.publish.MqttPublisher;
import top.hjcwzx.wrench.mqtt.strategy.MqttMessageHandler;
import top.hjcwzx.wrench.mqtt.strategy.MqttMessageManager;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 *
 * <p><b>类名：</b> MqttService</p>
 * <p><b>描述：</b> mqtt服务</p>
 * <p><b>创建时间：</b> 2025-06-17 15:34</p>
 *
 * @author 胡锦程
 */
public class MqttService implements MqttPublisher {

    public Logger log = LoggerFactory.getLogger(MqttService.class);
    private MqttClient client;
    private final MqttConnectOptions options = new MqttConnectOptions();
    private final MqttConfigProperties properties;

    private final MqttEventPublisher publisher;
    private final MqttMessageManager mqttMessageManager;

    private final ObjectMapper kitObjectMapper;
    private final Map<String, List<Consumer<Object>>> topicHandlerMap = new ConcurrentHashMap<>();
    private volatile boolean reconnecting = false;

    public MqttService(MqttConfigProperties properties, MqttEventPublisher publisher, MqttMessageManager mqttMessageManager, ObjectMapper kitObjectMapper) {
        this.properties = properties;
        this.publisher = publisher;
        this.mqttMessageManager = mqttMessageManager;
        this.kitObjectMapper = kitObjectMapper;
    }


    @PostConstruct
    public void init() {
        //使用自定义重连机制
        options.setAutomaticReconnect(false);
        options.setCleanSession(true);
        options.setUserName(properties.getUsername());
        options.setPassword(properties.getPassword().toCharArray());
    }

    public void connect() throws MqttException {
        String connectUrl = properties.getAgreement() + Constant.COLOR_WRITING_LINE + properties.getIp() + Constant.COLOR + properties.getPort();

        client = new MqttClient(connectUrl, properties.getClientId(), new MemoryPersistence());
        client.connect(options);
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverUri) {
                log.info("✅ MQTT连接成功 (reconnect={" + reconnect + "})");
                publisher.publish(new MqttConnectedEvent(this));
            }

            @Override
            public void connectionLost(Throwable cause) {
                log.warn("⚠️ MQTT连接丢失: {" + cause.getMessage() + "}");
                publisher.publish(new MqttConnectionLostEvent(this, cause));
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                MqttMessageHandler mqttMessageHandler = mqttMessageManager.getMqttMessageHandler(properties.getStrategyEnum());
                mqttMessageHandler.handleMessage(topic, payload, topicHandlerMap);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
            }
        });

    }

    /**
     * 订阅默认主题列表
     */
    public void subscribeDefaultTopics() {
        if (properties.getDefaultTopics() != null) {
            properties.getDefaultTopics().forEach(topic -> {
                try {
                    client.subscribe(topic);
                    log.info("订阅默认topic: {" + topic + "}");
                } catch (MqttException e) {
                    log.error("订阅默认topic失败: {}", topic, e);
                }
            });
        }
    }

    /**
     * 订阅注解主题列表
     */
    public void subscribeHandlers() {
        topicHandlerMap.keySet().forEach(topic -> {
            try {
                client.subscribe(topic);
                log.info("订阅注解topic: {" + topic + "}");
            } catch (MqttException e) {
                log.error("订阅注解topic失败: {}", topic, e);
            }
        });
    }

    /**
     * 泛型注册监听
     */
    public <T> void registerHandler(String topic, Class<T> type, Consumer<T> handler) {
        topicHandlerMap.computeIfAbsent(topic, k -> new ArrayList<>())
                .add(payload -> {
                    try {
                        T obj = type == String.class ? (T) payload : kitObjectMapper.readValue(payload.toString(), type);
                        handler.accept(obj);
                    } catch (Exception e) {
                        log.error("消息泛型转换失败: {}", e.getMessage());
                    }
                });
    }

    /**
     * 发送消息
     */
    @Override
    public void publish(String topic, Object message) throws MqttException, JsonProcessingException {
        if (client != null && client.isConnected()) {
            String payload = message instanceof String ? (String) message : kitObjectMapper.writeValueAsString(message);
            client.publish(topic, new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)));
        } else {
            throw new MqttException(new Throwable("MQTT未连接"));
        }
    }

    /**
     * 客户端重连
     */
    public synchronized boolean reconnect() {
        if (reconnecting) {
            return false;
        }
        reconnecting = true;

        try {
            // 若已有连接，先安全断开
            if (client != null && client.isConnected()) {
                try {
                    client.disconnect();
                    client.close();
                    log.info("🔌 MQTT 旧连接已断开，准备重连");
                } catch (Exception ex) {
                    log.warn("⚠️ MQTT 断开旧连接时异常: " + ex.getMessage());
                }
            }
            connect();
            // 如果连接成功，订阅主题
            if (client.isConnected()) {
                subscribeDefaultTopics();
                subscribeHandlers();
                log.info("✅ MQTT 重连成功并已订阅");
                return true;
            } else {
                log.error("❌ MQTT 重连失败，连接未建立");
                return false;
            }
        } catch (Exception e) {
            log.warn("MQTT 连接异常: {}", e);
            return false;
        } finally {
            reconnecting = false;
        }
    }
}
